package com.lagou.wwyan.rpc.register;

import com.lagou.wwyan.rpc.bean.RpcRequestMonitor;
import com.lagou.wwyan.rpc.listener.ServiceChangeListener;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

public class ZookeeperRpcRegister implements RpcRegister{
    private String url;
    private String RPC_PATH = "/rpc-wwyan/";
    private String MONIT_PATH = RPC_PATH + "monit";
    private List<ServiceChangeListener> listeners = new ArrayList<>();

    private CuratorFramework client;
    private Map<String,List<String>> serviceMap = new ConcurrentHashMap<>();

    private static final ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);

    public ZookeeperRpcRegister(String url) {
        this.url = url;

        client = CuratorFrameworkFactory.builder().connectString(url)
                .retryPolicy(new ExponentialBackoffRetry(1000,5))
                .build();
        client.getConnectionStateListenable().addListener((curatorFramework, connectionState) -> {
            if (ConnectionState.CONNECTED.equals(connectionState)){
                System.out.println("zookeeper连接成功！");
            }
        });

        client.start();


        //上报
        int delay = 5;
        executorService.scheduleWithFixedDelay(() -> {
            //上报
            List<RpcRequestMonitor.RequestTime> times = RpcRequestMonitor.getTimes();
            long millis = System.currentTimeMillis();
            for (RpcRequestMonitor.RequestTime time : times) {

                String path = MONIT_PATH + "/" + time.getHost() + ":" + time.getPort();
                if (millis - time.getStart() >= 5000){
                    String data = getData(path);
                    if (data == null || data.length() == 0){
                        //如果不存在5秒内的请求，则删除
                        time.setTimeLength(0L);
                        time.setStart(0L);
                        RpcRequestMonitor.update(time);
                        removeNode(path);
                    }else {
                        //更新本地的时间
                        String[] split = data.split(",");
                        Long start = Long.valueOf(split[0]);
                        if (start < 5000) {
                            time.setStart(start);
                            time.setTimeLength(Long.valueOf(split[1]));
                            RpcRequestMonitor.update(time);
                        }else {
                            //如果不存在5秒内的请求，则删除
                            time.setTimeLength(0L);
                            time.setStart(0L);
                            RpcRequestMonitor.update(time);
                            removeNode(path);
                        }
                    }

                }else {
//                    if (!exist(MONIT_PATH)){
//                        createNode(MONIT_PATH);
//                    }
                    //更新时长
                    updateData(path,time.getStart()+","+time.getTimeLength());

                }
            }

        },delay,delay, TimeUnit.SECONDS);


    }

    @Override
    public boolean registry(String service, String host, int port) {
        System.out.println("注册服务："+service);
        String path = providerPath(service);

        return createNode(host, port, path);
    }

    private boolean createNode(String host, int port, String path) {
        path = path + "/"+ host +":"+ port;
        return createNode(path);
    }

    private boolean createNode(String path) {
        try {
            client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(path);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }
    private boolean removeNode(String path) {
        try {
            if (exist(path)) {
                client.delete().deletingChildrenIfNeeded().forPath(path);
                return true;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }
    private boolean updateData(String path,String content) {
        try {
            byte[] bytes = content.getBytes();
            if (!exist(path)) {
                client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(path, bytes);
            }else {
                client.setData().forPath(path, bytes);
            }
            return true;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    public String getData(String path){
        try {
            if (!exist(path)) {
                return null;
            }else {
                byte[] bytes = client.getData().forPath(path);
                return new String(bytes);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public List<String> discovery(String service) {
        String path = providerPath(service);
        List<String> services = serviceMap.get(service);
        if (services == null || services.isEmpty()){
            try {
                services = client.getChildren().forPath(path);
                serviceMap.put(service,services);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        registryWatch(service,path);
        return services;
    }

    @Override
    public void destroy() {
        this.client.close();
    }

    @Override
    public void addListener(ServiceChangeListener listener) {
        listeners.add(listener);
    }

    private String providerPath(String service) {
        return RPC_PATH + service + "/provider";
    }


    public boolean exist(String path){
        try {
           return   client.checkExists().forPath(path) != null;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    private void registryWatch(final String service, final String path) {
        PathChildrenCache nodeCache = new PathChildrenCache(client, path, true);
        try {
            nodeCache.getListenable().addListener((client, pathChildrenCacheEvent) -> {
                // 更新本地的缓
                List<String> serviceList = client.getChildren().forPath(path);
                serviceMap.put(service,serviceList);
                listeners.stream().forEach(nodeChangeListener -> {
                    System.out.println("节点变化，开始通知业务");
                    PathChildrenCacheEvent.Type eventType = pathChildrenCacheEvent.getType();
                    System.out.println("收到节点变更通知:" + eventType + "------" + service + "---" + serviceList);
                    String childPath = pathChildrenCacheEvent.getData().getPath();
                    String instanceConfig = childPath.substring(childPath.lastIndexOf("/") + 1);

                    String[] address = instanceConfig.split(":");
                    String host = address[0];
                    int port = Integer.parseInt(address[1]);
                    // 增加节点
                    if (PathChildrenCacheEvent.Type.CHILD_ADDED.equals(eventType)
                            || PathChildrenCacheEvent.Type.CONNECTION_RECONNECTED.equals(eventType)) {

                        nodeChangeListener.serverAdd(service,host,port);
                        System.out.println("新增节点:" + instanceConfig);
                    } else if (PathChildrenCacheEvent.Type.CHILD_REMOVED.equals(eventType)
                            || PathChildrenCacheEvent.Type.CONNECTION_SUSPENDED.equals(eventType)
                            || PathChildrenCacheEvent.Type.CONNECTION_LOST.equals(eventType)) {
                        // 移除节点
                        nodeChangeListener.serverRemove(service,host,port);
                    }

                });
            });
            nodeCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
